#ifndef __LOPPER_HPP__
#define __LOPPER_HPP__

#include "Buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <thread>
#include <memory>
#include <atomic>

namespace MyLog
{
    using Functor =std::function<void (Buffer&)>;
    enum class AsyncType
    {
        ASYNC_SAFE,//安全状态，不会无限扩容
        ASYNC_UNSAFE//
    };
    class AsyncLooper
    {
    public:
        using Asyncshptr=std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &func,const AsyncType &lopper=AsyncType::ASYNC_SAFE)
        :_loppertype(lopper),
         _stop(false),
         _thread(std::thread(&AsyncLooper::threadEntry,this)),
         _callback(func)
        {}
        ~AsyncLooper(){stop();}
        void stop()
        {
            _stop = true;
            //异步线程停止后就唤醒业务线程进行工作
            // _cond_product.notify_all();
            _cond_consumer.notify_all();//唤醒所有消费者线程
            // if (_thread.joinable())
                _thread.join();
        }
        //生产缓冲区
        void Push(const char *data,std::size_t len)
        {
            // 1.无限扩容-不安全
            // 2.固定大小--生产缓冲区满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            //条件变量控制，如果缓冲区剩余空间大小大于数据长度，就可以继续添加数据
            //处于安全状态缓冲区空间不够了就阻塞
            if (_product_buf.writeAbleSize() < len) 
            {
                _product_buf.ensureEnoughSpace(len);
            }
            if(_loppertype == AsyncType::ASYNC_SAFE)
            {
                _cond_product.wait(lock,[&](){
                    return _product_buf.writeAbleSize() >= len;
                });
            }
            //到这里说明缓冲区空间足够
            _product_buf.push(data, len);
            //唤醒消费者线程处理，只有一个异步线程，唤醒一个和全部唤醒是一样的
            _cond_consumer.notify_one();
        }
    private:
        //线程入口函数
        // 对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
        void threadEntry()
        {
            // 为互斥锁设置一个生命周期，当缓冲区交换完毕后就解锁（并不对数据的处理过程做加锁保护）
            // 让锁的颗粒度尽可能的细
            while(true)
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 根据生产缓冲区中是否有数据来判断交换还是阻塞
                    _cond_consumer.wait(lock,[&](){
                        return (_stop || !_product_buf.empty());
                    });
                    if(_stop && _product_buf.empty())break;
                    // 走到这里说明生产缓冲区里有数据了，交换缓冲区让异步线程去处理
                    _consumer_buf.swap(_product_buf);
                    //交换后生产缓冲区又有空间了，唤醒业务线程继续生产
                    //安全情况下缓冲区空间不够了不需要唤醒线程
                    if(_loppertype == AsyncType::ASYNC_SAFE)
                        _cond_product.notify_all();
                }
                // 异步线程处理消费缓冲区的数据
                _callback(_consumer_buf);
                // 处理完成后重置读写指针的位置
                _consumer_buf.reset();
            }
            
        }
    private:
        Functor _callback;//处理消费者缓冲区数据的回调函数
    private:
        AsyncType _loppertype;
        Buffer _product_buf;//生产者缓冲区
        Buffer _consumer_buf;//消费者缓冲区
        std::atomic<bool> _stop;//异步线程是否停止
        std::mutex _mutex;//互斥锁
        std::thread _thread;//异步工作线程
        std::condition_variable _cond_product;
        std::condition_variable _cond_consumer;
    };
}

#endif